package com.jy.oms.framework.asyn.task.queue.redis;

import com.alibaba.fastjson.JSON;
import com.jy.oms.framework.asyn.task.checker.model.ExecutorCheckResult;
import com.jy.oms.framework.asyn.task.constants.CacheConstants;
import com.jy.oms.framework.asyn.task.constants.TaskQueueTypeEnum;
import com.jy.oms.framework.asyn.task.constants.TaskStateEnum;
import com.jy.oms.framework.asyn.task.core.RedissonClientProxy;
import com.jy.oms.framework.asyn.task.handler.ITaskHandlerManager;
import com.jy.oms.framework.asyn.task.handler.ITaskHandlerProcessor;
import com.jy.oms.framework.asyn.task.model.*;
import com.jy.oms.framework.asyn.task.queue.AbstractTaskQueueManager;
import com.jy.oms.framework.asyn.task.queue.redis.bo.RedisTaskBo;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RBucket;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

@Service
public class RedisTaskQueueManager extends AbstractTaskQueueManager implements ITaskHandlerProcessor, DisposableBean {
    private static Logger logger = LoggerFactory.getLogger(RedisTaskQueueManager.class);
    @Autowired
    private ITaskHandlerManager taskHandlerManager;
    private Map<String, RDelayedQueue<String>> delayedQueueMap = new ConcurrentHashMap<>();

    @Override
    public void processAfterTaskHandlerInitialization() {
        Map<TaskGroup, List<TaskHandlerWrapper>> taskGroupListMap = taskHandlerManager.loadAllExecutableTaskHandler();
        if (MapUtils.isEmpty(taskGroupListMap)) {
            return;
        }

        //提前初始化延迟队列有意义
        //[
        // 如果是在添加任务初始化延迟队列，当应用重启时，且长时间没有添加任务，那么已存在的待过期的任务不会被执行，数据不会被迁移到堵塞队列上；
        // 需要了解redisson的延迟队列的实现
        //]
        taskGroupListMap.keySet().stream()
                .filter(taskGroup -> taskGroup.getTaskQueueType() == TaskQueueTypeEnum.REDIS)
                .map(TaskGroup::getGroup)
                .forEach(group -> {
                    delayedQueueMap.computeIfAbsent(group, this::getRDelayedQueue);
                });
    }

    @Override
    public void destroy() throws Exception {
        delayedQueueMap.entrySet().forEach(entry -> {
            entry.getValue().destroy();
        });
    }

    private RDelayedQueue<String> getRDelayedQueue(String taskGroup) {
        RedissonClient redissonClient = RedissonClientProxy.getRedissonClient();
        RBlockingQueue<String> queue = redissonClient.getBlockingQueue(this.getQueueKey(taskGroup));
        RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(queue);
        return delayedQueue;
    }

    private String getQueueKey(String taskGroup) {
        return CacheConstants.genCacheKey(CacheConstants.TASK_QUEUE, taskGroup);
    }

    private String getParamQueueKey(String taskGroup, String paramIdentifier) {
        return CacheConstants.genCacheKey(CacheConstants.TASK_QUEUE, taskGroup, paramIdentifier);
    }

    @Override
    public void add(TaskWrapper taskWrapper) {
        super.add(taskWrapper);

        TaskDto taskDto = taskWrapper.getTaskDto();
        if (taskDto.getDelayedSeconds() == null) {
            taskDto.setDelayedSeconds(1L);
        }

        RedisTaskBo redisTaskBo = new RedisTaskBo();
        redisTaskBo.setTaskDto(taskDto);
        redisTaskBo.setCallCount(taskWrapper.getCallCount());
        if (StringUtils.isBlank(taskWrapper.getParamIdentifier())) {
            String taskParam = this.convertTaskParam(taskDto);
            redisTaskBo.setParamIdentifier(this.generateTaskIdentifier(taskDto.getTaskCode(), taskParam));
        } else {
            redisTaskBo.setParamIdentifier(taskWrapper.getParamIdentifier());
        }

        RedissonClient redissonClient = RedissonClientProxy.getRedissonClient();
        String paramIdentifier = redisTaskBo.getParamIdentifier();
        RDelayedQueue<String> delayedQueue;
        if (null == (delayedQueue = (delayedQueueMap.get(taskWrapper.getTaskGroup())))) {
            delayedQueue = delayedQueueMap.computeIfAbsent(taskWrapper.getTaskGroup(), this::getRDelayedQueue);
        }
        delayedQueue.offer(paramIdentifier, taskDto.getDelayedSeconds(), TimeUnit.SECONDS);

        //延迟堵塞队列任务到时由于某种原因长时间未执行，这种场景下，如果过期时间不足1小时（过短），任务数据有丢失风险，
        //因此将过短的延时时间按照1小时计算
        long delayedSeconds = taskDto.getDelayedSeconds() > 3600 ? taskDto.getDelayedSeconds() : 3600;
        RBucket<String> bucket = redissonClient.getBucket(this.getParamQueueKey(taskWrapper.getTaskGroup(), paramIdentifier));
        bucket.set(JSON.toJSONString(redisTaskBo), delayedSeconds * 10, TimeUnit.SECONDS);
    }

    @Override
    public void addBatch(List<TaskWrapper> taskWrapperList) {
        for (TaskWrapper taskWrapper : taskWrapperList) {
            this.add(taskWrapper);
        }
    }

    @Override
    public List<TaskWrapper> poll(TaskQueryDto taskQueryDto) {
        List<TaskWrapper> taskWrapperList = new ArrayList<>();
        RedissonClient redissonClient = RedissonClientProxy.getRedissonClient();
        RBlockingQueue<String> queue = redissonClient.getBlockingQueue(this.getQueueKey(taskQueryDto.getTaskGroup()));

        String paramIdentifier = null;
        try {
            //等待队列有数据，否则一直堵塞
            paramIdentifier = queue.take();
        } catch (InterruptedException e) {
            logger.warn("队列阻塞中断", e);
        }

        if (StringUtils.isBlank(paramIdentifier)) {
            return null;
        }

        RBucket<String> bucket = redissonClient.getBucket(this.getParamQueueKey(taskQueryDto.getTaskGroup(), paramIdentifier));
        String taskJson = bucket.getAndDelete();
        if (StringUtils.isBlank(taskJson)) {
            return null;
        }

        RedisTaskBo redisTaskBo = JSON.parseObject(taskJson, RedisTaskBo.class);
        TaskDto taskDto = redisTaskBo.getTaskDto();
        Object param = this.getTaskParam(taskDto.getTaskCode(), taskDto.getParam().toString(), redisTaskBo.getParamIdentifier());
        if (param == null) {
            return null;
        }
        taskDto.setParam(param);

        TaskWrapper taskWrapper = new TaskWrapper();
        taskWrapper.setTaskCode(taskDto.getTaskCode());
        taskWrapper.setTaskDto(taskDto);
        taskWrapper.setTaskGroup(taskQueryDto.getTaskGroup());
        taskWrapper.setCallCount(redisTaskBo.getCallCount());
        taskWrapper.setParamIdentifier(redisTaskBo.getParamIdentifier());
        taskWrapperList.add(taskWrapper);

        return taskWrapperList;
    }

    @Override
    public void finish(TaskResultDto taskResultDto) {
        super.finish(taskResultDto);

        if (TaskStateEnum.DONE == taskResultDto.getTaskState()) {
            return;
        }

        TaskWrapper taskWrapper = taskResultDto.getTaskWrapper();
        TaskDto taskDto = taskWrapper.getTaskDto();
        long delayedSeconds = 0;
        if (taskResultDto.isCheckFilter()) {
            ExecutorCheckResult executorCheckResult = taskResultDto.getExecutorCheckResult();
            if (executorCheckResult.getEarliestExecuteTime() != null) {
                long gap = executorCheckResult.getEarliestExecuteTime().getTime() - System.currentTimeMillis();
                delayedSeconds = (int) gap / 1000;
            } else if (TaskStateEnum.DONE == executorCheckResult.getTaskState()) {
                //do nothing
            } else {
                //检查不通过情况下，默认延时10分钟再次查询检查任务是否可执行
                delayedSeconds = 10 * 60;
            }
        } else if (taskResultDto.getTaskState() == TaskStateEnum.INVALID) {
            //无效任务延时1天
            delayedSeconds = 24 * 60 * 60;
        } else if (taskResultDto.getTaskState() == TaskStateEnum.TIMEOUT) {
            //超时任务延时1小时(超时并不是没执行成功，延时时间不能小于[TaskRunningStateCollector]运行状态过期时间)
            delayedSeconds = 61 * 60;

            this.alarmTimeoutTask(taskResultDto);
        } else if (taskResultDto.isEnableRetry()) {
            //重试任务
            if(taskResultDto.getRetryIntervalSeconds() == 0){
                delayedSeconds = taskResultDto.getCallCount();
            }else{
                delayedSeconds = taskResultDto.isRetryBackoff()
                        ? (long) (Math.min(taskResultDto.getCallCount(), Math.max(taskResultDto.getRetryNumConfig(), 1))) * taskResultDto.getRetryIntervalSeconds()
                        : taskResultDto.getRetryIntervalSeconds();
            }
        }

        taskWrapper.setCallCount(taskResultDto.getCallCount());
        if (delayedSeconds > 0) {
            //任务重进入队列
            taskDto.setDelayedSeconds(delayedSeconds);
            this.add(taskWrapper);
        }
    }

    @Override
    public void finish(List<TaskResultDto> taskResultList) {
        for (TaskResultDto taskResultDto : taskResultList) {
            this.finish(taskResultDto);
        }
    }

    @Override
    public void remove(TaskWrapper taskWrapper) {
        TaskDto taskDto = taskWrapper.getTaskDto();
        String taskParam = this.convertTaskParam(taskDto);
        String paramIdentifier = this.generateTaskIdentifier(taskDto.getTaskCode(), taskParam);
        logger.info("删除redis任务, taskGroup={}, taskCode={}, paramIdentifier={}", taskWrapper.getTaskGroup(), taskDto.getTaskCode(), paramIdentifier);

        RedissonClient redissonClient = RedissonClientProxy.getRedissonClient();
        RBlockingQueue<String> queue = redissonClient.getBlockingQueue(this.getQueueKey(taskWrapper.getTaskGroup()));
        RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(queue);
        delayedQueue.remove(paramIdentifier);

        RBucket<String> bucket = redissonClient.getBucket(this.getParamQueueKey(taskWrapper.getTaskGroup(), paramIdentifier));
        bucket.delete();
    }

}
